/*
    Copyright 2009 Spencer B. Koehler, Ph.D.

    This file is part of the Brillig Toolkit.

    The Brillig Toolkit is free software: you can redistribute it and/or modify
    it under the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    The Brillig Toolkit is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public License
    along with The Brillig Toolkit.  If not, see <http://www.gnu.org/licenses/>.
*/
package org.brillig.cluster.job;

import org.brillig.cio.MessageHelper;
import org.brillig.cluster.config.ClusterException;
import org.brillig.cluster.config.Config;
import org.brillig.util.PropertiesParser;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.util.Date;
import java.util.Properties;

/**
 * A job for serving work to other jobs across a cluster.
 * <p>
 * @author Spence Koehler
 */
public class BatchWorkServer extends AbstractBatchWorkServer {
  private String batchFileName;

  public BatchWorkServer() {
    super();
  }

  public BatchWorkServer(Properties properties) {
    super(properties);

    this.batchFileName = properties.getProperty("batch");

    if (batchFileName == null) {
      throw new IllegalArgumentException("Missing required property 'batch'!");
    }
  }

  public String getBatchFilename() {
    return batchFileName;
  }

  /**
   * Write thie message to the dataOutput stream such that this message
   * can be completely reconstructed through this.read(dataInput).
   *
   * @param dataOutput  the data output to write to.
   */
  public void write(DataOutput dataOutput) throws IOException {
    super.write(dataOutput);
    MessageHelper.writeString(dataOutput, batchFileName);
  }

  /**
   * Read this message's contents from the dataInput stream that was written by
   * this.write(dataOutput).
   * <p>
   * NOTE: this requires all implementing classes to have a default constructor
   *       with no args.
   *
   * @param dataInput  the data output to write to.
   */
  public void read(DataInput dataInput) throws IOException {
    super.read(dataInput);
    this.batchFileName = MessageHelper.readString(dataInput);
  }

  protected final String buildOutputFilename() {
    final Config config = getConfig();
    return config.getOutputDataPath(getName(), getJobId(), batchFileName);
  }
  
  protected PathBatch getPathBatch() {
    final Config config = getConfig();

    // jvmRootDir/data/job/WorkServer/<workServerId>/<batchFileName>
    final File workDir = new File(config.getJobDataPath(getName(), getJobId()));
    final File pathBatchFile = new File(workDir, batchFileName);
    PathBatch result = null;

    if (!pathBatchFile.exists()) {
      System.err.println(new Date() + ": *** Can't find path batch file at '" + pathBatchFile.getAbsolutePath() + "'!");
    }
    else {
      try {
        result = new PathBatch(pathBatchFile, onlyOwn, acceptHelp);

        System.out.println(new Date() + ": Opened pathBatch at '" + pathBatchFile.getAbsolutePath() + "'. " +
                           getDescription());
      }
      catch (IOException e) {
        System.err.println(new Date() + ": *** Can't load path batch file at '" + pathBatchFile.getAbsolutePath() + "'!");
        e.printStackTrace(System.err);
      }
    }
    return result;
  }

  public static void main(String[] args) throws IOException, ClusterException {
    // start the work server on a node
    //
    //properties: [user], [defName], [machines], jobId(=dataDirName), groupName(=workServerId), batch, [onlyOwn] [acceptHelp] [limit]
    final PropertiesParser pp = new PropertiesParser(args);
    final WorkServer workServer = new BatchWorkServer(pp.getProperties());
    run(pp, workServer);
  }
}
